import sys
from inspect import getdoc
from itertools import chain
from os import PathLike
from string import Template
from typing import Callable, Literal, Optional, Union

from anyio import Path, create_task_group, run_process, to_thread
from pydantic import BaseModel
from pydantic.json_schema import JsonSchemaMode, JsonSchemaValue, models_json_schema
from pydantic_core import to_json
from typing_extensions import NamedTuple, TypeIs

if sys.version_info < (3, 10):
    NoneType = type(None)
else:
    from types import NoneType

__all__ = ["gen_ts"]


_COMMANDS_TITLE = "Commands"
"""Default title for commands JSON schema."""

_INPUT_PROPERTIES = "input"
_OUTPUT_PROPERTIES = "output"

# Ref: <https://github.com/bcherny/json-schema-to-typescript/pull/168>
_VOID_TS_TYPE = {
    "tsType": "void | undefined",
}
_BYTES_TS_TYPE = {
    "tsType": "ArrayBuffer",
}

_API_TYPES_FILE_NAME = "_apiTypes.d.ts"
"""Default file name for generated TypeScript types for API."""

_API_CLIENT_FILE_NAME = "apiClient.ts"
"""Default file name for generated TypeScript API client."""

_CLIENT_PREFIX = f"""\
/* eslint-disable */
/**
 * This file was automatically generated by pytauri-gen-ts.
 * DO NOT MODIFY IT BY HAND. Instead, modify the source commands API,
 * and run pytauri-gen-ts to regenerate this file.
 */

import {{ pyInvoke }} from "tauri-plugin-pytauri-api";
import type {{ InvokeOptions }} from "@tauri-apps/api/core";

import type {{ {_COMMANDS_TITLE} }} from "./{_API_TYPES_FILE_NAME}";

"""

_INVOKE_CMD_TEMPLATE = Template(f"""\
${{doc}}export async function ${{js_cmd}}(
    body: {_COMMANDS_TITLE}["${{py_cmd}}"]["{_INPUT_PROPERTIES}"],
    options?: InvokeOptions
): Promise<{_COMMANDS_TITLE}["${{py_cmd}}"]["{_OUTPUT_PROPERTIES}"]> {{
    return await pyInvoke("${{py_cmd}}", body, options);
}}
""")


_AliasGenerator = Callable[[str], str]

_Model = type[BaseModel]
_Bytes = type[bytes]
_Void = type[None]


class InputOutput(NamedTuple):
    cmd_handler: object
    input_type: Union[_Model, _Bytes, _Void]
    output_type: Union[_Model, _Bytes, _Void]


CommandInputOutput = dict[str, InputOutput]


class _InputOutputSerde(NamedTuple):
    input_type: Union[
        tuple[_Model, Literal["validation"]],
        _Bytes,
        _Void,
    ]
    output_type: Union[
        tuple[_Model, Literal["serialization"]],
        _Bytes,
        _Void,
    ]


_CommandInputOutputSerde = dict[str, _InputOutputSerde]


def _is_model(
    type_: Union[_Model, _Bytes, _Void],
) -> TypeIs[_Model]:
    return issubclass(type_, BaseModel)


def _is_model_serde(
    type_: Union[tuple[_Model, JsonSchemaMode], _Bytes, _Void],
) -> TypeIs[tuple[_Model, JsonSchemaMode]]:
    return isinstance(type_, tuple)


# -- core ---------------------------------------------------------------------


def _gen_json_schemas(cmd_in_out: CommandInputOutput) -> JsonSchemaValue:
    cmd_in_out_serde: _CommandInputOutputSerde = {
        cmd: _InputOutputSerde(
            input_type=(input_type, "validation")
            if _is_model(input_type)
            else input_type,
            output_type=(output_type, "serialization")
            if _is_model(output_type)
            else output_type,
        )
        for cmd, (_, input_type, output_type) in cmd_in_out.items()
    }

    json_schemas_map, definitions = models_json_schema(
        tuple(filter(_is_model_serde, chain.from_iterable(cmd_in_out_serde.values()))),
        title=_COMMANDS_TITLE,
        description="Commands Input and Output Schemas",
    )

    json_schemas = {
        **_object_json_schema(
            {
                cmd: _object_json_schema(
                    {
                        _INPUT_PROPERTIES: json_schemas_map[input_type]
                        if _is_model_serde(input_type)
                        else _VOID_TS_TYPE
                        if issubclass(input_type, NoneType)
                        else _BYTES_TS_TYPE,
                        _OUTPUT_PROPERTIES: json_schemas_map[output_type]
                        if _is_model_serde(output_type)
                        else _VOID_TS_TYPE
                        if issubclass(output_type, NoneType)
                        else _BYTES_TS_TYPE,
                    }
                )
                for cmd, (input_type, output_type) in cmd_in_out_serde.items()
            }
        ),
        **definitions,  # $defs, title, description
    }

    return json_schemas


def _gen_json_schemas_bytes(cmd_in_out: CommandInputOutput) -> bytes:
    json_schemas = _gen_json_schemas(cmd_in_out)
    return to_json(json_schemas)


def _gen_client_code(
    cmd_in_out: CommandInputOutput, *, cmd_alias: Optional[_AliasGenerator] = None
) -> str:
    return _CLIENT_PREFIX + "\n".join(
        _INVOKE_CMD_TEMPLATE.substitute(
            doc=_gen_ts_doc(in_out.cmd_handler) or "",
            js_cmd=cmd_alias(cmd) if cmd_alias else cmd,
            py_cmd=cmd,
        )
        for cmd, in_out in cmd_in_out.items()
    )


async def gen_ts(
    cmd_in_out: CommandInputOutput,
    output_dir: Union[str, PathLike[str]],
    json2ts_cli: str,
    *,
    cmd_alias: Optional[_AliasGenerator] = None,
):
    output_dir_path = Path(output_dir)
    await output_dir_path.mkdir(parents=True, exist_ok=True)

    async def gen_api_types():
        json_schemas_bytes = await to_thread.run_sync(
            _gen_json_schemas_bytes,
            cmd_in_out,
        )
        process = await run_process(json2ts_cli, input=json_schemas_bytes, check=True)

        api_types_file_path = output_dir_path / _API_TYPES_FILE_NAME
        await _write_if_changed_bytes(api_types_file_path, process.stdout)

    async def gen_api_client():
        api_client_code = _gen_client_code(cmd_in_out, cmd_alias=cmd_alias)

        api_client_file_path = output_dir_path / _API_CLIENT_FILE_NAME
        await _write_if_changed_text(api_client_file_path, api_client_code)

    async with create_task_group() as tg:
        tg.start_soon(gen_api_types)
        tg.start_soon(gen_api_client)


# -- utils --------------------------------------------------------------------


def _object_json_schema(properties: dict[str, JsonSchemaValue]) -> JsonSchemaValue:
    return {
        "type": "object",
        "properties": properties,
        "required": list(properties.keys()),
        "additionalProperties": False,
    }


# TODO: use stat to cache and compare in chunks,
# ref: <https://github.com/python/cpython/blob/a2f1d22a362cccab59691fa4fa2b202475480b04/Lib/filecmp.py#L30-L68>
async def _write_if_changed_text(
    path: Path,
    content: str,
) -> None:
    encoding = "utf-8"
    try:
        if await path.read_text(encoding=encoding) == content:
            return
    except FileNotFoundError:
        pass

    await path.write_text(content, encoding=encoding)


async def _write_if_changed_bytes(
    path: Path,
    content: bytes,
) -> None:
    try:
        if await path.read_bytes() == content:
            return
    except FileNotFoundError:
        pass

    await path.write_bytes(content)


def _gen_ts_doc(obj: object) -> Optional[str]:
    py_doc = getdoc(obj)
    if not py_doc:  # Skip generation for empty docstrings
        return None

    # `indent`, ref: <https://github.com/python/cpython/blob/6d21cc54fffbe56df3372f65227160bf27807158/Lib/textwrap.py#L447-L469>
    js_doc_body_lines: list[str] = []
    for line in py_doc.splitlines(keepends=True):
        # str.splitlines(keepends=True) doesn't produce the empty string,
        # so we need to use `str.isspace()` rather than a truth test.
        if line.isspace():
            js_doc_body_lines.append(" *\n")
        else:
            js_doc_body_lines.extend((" * ", line))

    js_doc_body = "".join(js_doc_body_lines)

    return f"""\
/**
{js_doc_body}
 */
"""
